package com.chb.flink.time

import java.text.SimpleDateFormat
import java.util.Date

import com.chb.flink.source.{MyCustomerSource, StationLog}
import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * 每隔5秒统计一下最近10秒内，每个基站中通话时间最长的一次通话发生的时间还有，
 * 主叫号码，被叫号码，通话时长，并且还得告诉我们当前发生的时间范围（10秒）
 */
object MaxCallTime {

    def main(args: Array[String]): Unit = {
        val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        import org.apache.flink.streaming.api.scala._
        streamEnv.setParallelism(1)


        val data = streamEnv.addSource(new MyCustomerSource)
            .assignAscendingTimestamps(_.callTime) // 引入watermark, 通过参数指定eventTime
            .filter(_.callType == "success")
        // 分组，开窗
        data.keyBy(_.sid)
            .timeWindow(Time.seconds(10), Time.seconds(5))
            .reduce(new MyReduceFunction(), new ReturnMaxTimeWindowFunction)
            .print()


        streamEnv.execute()

    }

    class MyReduceFunction extends ReduceFunction[StationLog] {
        override def reduce(t: StationLog, t1: StationLog): StationLog = {
            // 比较通话时长最长的
            if (t.duration > t1.duration) t else t1
        }
    }

}